package com.atguigu.day01;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class Flink01_Batch_WordCount {
    public static void main(String[] args) throws Exception {
        /**
         * sparkCore中计算WordCount
         * 1.获取环境
         * 2.读取文件
         * 3.flatMap切出每一个单词
         * 4.Map将每一个单词组成Tuple2元组
         * 5.reduceByKey（1.将相同单词的数据聚合到一块2.根据传入的函数做累加计算）
         * 6.打印到控制台
         */

        //1.获取批处理环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

//         2.读取文件
        DataSource<String> dataSource = env.readTextFile("input/word.txt");

//         3.flatMap切出每一个单词
        FlatMapOperator<String, String> word = dataSource.flatMap(new MyFlatMap());

//         4.Map将每一个单词组成Tuple2元组
//        MapOperator<String, Tuple2<String, Integer>> wordToOne = word.map( value -> Tuple2.of(value, 1)).returns(Types.TUPLE(Types.STRING,Types.INT));
        MapOperator<String, Tuple2<String, Integer>> wordToOne = word.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value,1);
            }
        });

//         reduceByKey（1.将相同单词的数据聚合到一块2.根据传入的函数做累加计算）
        //5.将相同单词聚合到一块
        UnsortedGrouping<Tuple2<String, Integer>> groupBy = wordToOne.groupBy(0);

        //5.2做累加计算
        AggregateOperator<Tuple2<String, Integer>> result = groupBy.sum(1);

//         6.打印到控制台
        result.print();
    }

    public static class MyFlatMap implements FlatMapFunction<String,String>{

        @Override
        public void flatMap(String value, Collector<String> out) throws Exception {
            //1.将数据按照空格切分
            String[] words = value.split(" ");

            //2.遍历字符串数组，获取到每一个单词
            for (String word : words) {
                //将每一个单词发送至下游
                out.collect(word);
            }
        }
    }
}
